package fun.tan90.easy.binlog.worker.handler.processor;


import cn.hutool.core.date.DateUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.github.benmanes.caffeine.cache.LoadingCache;
import com.github.shyiko.mysql.binlog.event.deserialization.json.JsonBinary;
import fun.tan90.easy.binlog.base.constants.EasyBinlogConstants;
import fun.tan90.easy.binlog.base.constants.EventTypeConstants;
import fun.tan90.easy.binlog.base.enums.OutputResEnum;
import fun.tan90.easy.binlog.base.model.HeartBeat;
import fun.tan90.easy.binlog.base.model.OutputCount;
import fun.tan90.easy.binlog.base.model.RotateBinlogFile;
import fun.tan90.easy.binlog.plugin.common.constants.DamiEventConstants;
import fun.tan90.easy.binlog.plugin.common.event.BinlogDataEvent;
import fun.tan90.easy.binlog.plugin.common.event.BinlogDataEventRes;
import fun.tan90.easy.binlog.plugin.common.model.BinlogData;
import fun.tan90.easy.binlog.plugin.common.model.BinlogOutputRule;
import fun.tan90.easy.binlog.plugin.common.model.TableMeta;
import fun.tan90.easy.binlog.plugin.common.model.TableMetaParam;
import fun.tan90.easy.binlog.worker.client.EbWorker;
import fun.tan90.easy.binlog.worker.model.BinlogTableMap;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.noear.dami.Dami;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import javax.annotation.Resource;
import java.io.Serializable;
import java.util.*;
import java.util.stream.Collectors;

/**
 * description TODO
 * company 铁人科技
 *
 * @author 陈浩杰
 * @date 2023/9/1 12:53
 */
@Slf4j
@Component
public class BinlogEventProcessor {

    @Resource
    EbWorker ebWorker;

    @Resource
    LoadingCache<TableMetaParam, TableMeta> binlogColumnsCache;

    @Resource
    LoadingCache<String, Map<String, List<BinlogOutputRule>>> outputRuleCache;

    public void handleWrite(long timestamp, String masterCode, Integer offsetMilliseconds, BinlogTableMap binlogTableMap, List<Serializable[]> data) {
        TableMeta tableMeta = handleBefore(masterCode, binlogTableMap.getDatabase(), binlogTableMap.getTable());
        List<String> columnNames = tableMeta.getColumnNames();
        List<String> dataTypes = tableMeta.getDataTypes();
        List<BinlogData> binlogDataList = new ArrayList<>();
        for (Serializable[] item : data) {
            log.info("data size {}, column size {}", item.length, columnNames.size());
            JSONObject property = new JSONObject();
            for (int i = 0; i < columnNames.size(); i++) {
                String columnName = columnNames.get(i);
                property.putOnce(columnName, transformData(dataTypes.get(i), item[i], offsetMilliseconds));
            }
            BinlogData binlogData = BinlogData.builder()
                    .timestamp(timestamp)
                    .eventType(EventTypeConstants.WRITE)
                    .database(binlogTableMap.getDatabase())
                    .table(binlogTableMap.getTable())
                    .data(property)
                    .build();
            binlogDataList.add(binlogData);
        }
        handleAfter(EventTypeConstants.WRITE, masterCode, tableMeta, binlogDataList);
    }

    public void handleUpdate(long timestamp, String masterCode, Integer offsetMilliseconds, BinlogTableMap binlogTableMap, List<Map.Entry<Serializable[], Serializable[]>> data) {
        TableMeta tableMeta = handleBefore(masterCode, binlogTableMap.getDatabase(), binlogTableMap.getTable());
        List<String> columnNames = tableMeta.getColumnNames();
        List<String> dataTypes = tableMeta.getDataTypes();
        List<BinlogData> binlogDataList = new ArrayList<>();
        for (Map.Entry<Serializable[], Serializable[]> item : data) {
            JSONObject property = new JSONObject();
            JSONObject originalProperty = new JSONObject();
            Serializable[] key = item.getKey();
            Serializable[] value = item.getValue();
            for (int i = 0; i < columnNames.size(); i++) {
                String dataType = dataTypes.get(i);
                String columnName = columnNames.get(i);
                originalProperty.putOnce(columnName, transformData(dataType, key[i], offsetMilliseconds));
                property.putOnce(columnName, transformData(dataType, value[i], offsetMilliseconds));
            }
            BinlogData binlogData = BinlogData.builder()
                    .timestamp(timestamp)
                    .eventType(EventTypeConstants.UPDATE)
                    .database(binlogTableMap.getDatabase())
                    .table(binlogTableMap.getTable())
                    .data(property)
                    .originalData(originalProperty)
                    .build();
            // 比对出修改列
            List<String> diffCols = property.keySet().stream().filter(n -> !property.get(n).equals(originalProperty.get(n))).collect(Collectors.toList());
            binlogData.setDiffCols(diffCols);

            binlogDataList.add(binlogData);
        }
        handleAfter(EventTypeConstants.UPDATE, masterCode, tableMeta, binlogDataList);
    }

    public void handleDelete(long timestamp, String masterCode, Integer offsetMilliseconds, BinlogTableMap binlogTableMap, List<Serializable[]> data) {
        TableMeta tableMeta = handleBefore(masterCode, binlogTableMap.getDatabase(), binlogTableMap.getTable());
        List<String> columnNames = tableMeta.getColumnNames();
        List<String> dataTypes = tableMeta.getDataTypes();
        List<BinlogData> binlogDataList = new ArrayList<>();
        for (Serializable[] item : data) {
            JSONObject property = new JSONObject();
            for (int i = 0; i < columnNames.size(); i++) {
                String columnName = columnNames.get(i);
                property.putOnce(columnName, transformData(dataTypes.get(i), item[i], offsetMilliseconds));
            }
            BinlogData binlogData = BinlogData.builder()
                    .timestamp(timestamp)
                    .eventType(EventTypeConstants.DELETE)
                    .database(binlogTableMap.getDatabase())
                    .table(binlogTableMap.getTable())
                    .originalData(property)
                    .build();
            binlogDataList.add(binlogData);
        }
        handleAfter(EventTypeConstants.DELETE, masterCode, tableMeta, binlogDataList);
    }

    private TableMeta handleBefore(String masterCode, String database, String table) {
        log.info("masterCode {}，database {}, table {}", masterCode, database, table);
        return binlogColumnsCache.get(TableMetaParam.builder().masterCode(masterCode).database(database).table(table).build());
    }

    private void handleAfter(String eventType, String masterCode, TableMeta tableMeta, List<BinlogData> binlogDataList) {
        log.info("eventType {} tableMeta {}", eventType, tableMeta);
        Map<String, List<BinlogOutputRule>> groupedOutputRuleParam = outputRuleCache.get(masterCode);

        if (CollectionUtils.isEmpty(groupedOutputRuleParam)) {
            log.info("暂无转发规则 {}", masterCode);
            return;
        }

        BinlogDataEvent binlogDataEvent = BinlogDataEvent.builder()
                .eventType(eventType)
                .masterCode(masterCode)
                .tableMeta(tableMeta)
                .binlogDataList(binlogDataList)
                .build();

        for (Map.Entry<String, List<BinlogOutputRule>> entry : groupedOutputRuleParam.entrySet()) {
            String outputType = entry.getKey();
            binlogDataEvent.setBinlogOutputRules(entry.getValue());
            Dami.<BinlogDataEvent, BinlogDataEventRes>bus().sendAndSubscribe(DamiEventConstants.BINLOG_EVENT_TOPIC + outputType, binlogDataEvent, res -> {
                String outputCountId = res.getOutputCountId();
                OutputResEnum outputRes = res.getOutputRes();
                log.info("binlog_output_rule_count ID:{} {} 转发{} ", outputCountId, outputType, outputRes);
                outputCount(outputCountId, outputRes);
            });
        }

    }

    @SneakyThrows
    private Serializable transformData(String dataType, Serializable serializable, Integer offsetMilliseconds) {
        if (Objects.isNull(serializable)) {
            return null;
        }
        if ("datetime".equals(dataType)) {
            serializable = DateUtil.offsetMillisecond((Date) serializable, offsetMilliseconds);
            return serializable;
        }
        if ("json".equals(dataType)) {
            return JsonBinary.parseAsString((byte[]) serializable);
        }
        return serializable;
    }

    public void rotateBinlogFile(String masterCode, String binlogFilename, long binlogPosition) {
        RotateBinlogFile rotateBinlogFile = RotateBinlogFile.builder().binlogFilename(binlogFilename).binlogPosition(binlogPosition).masterCode(masterCode).build();
        ebWorker.send(EasyBinlogConstants.ROTATE_BINLOG_FILE, JSONUtil.toJsonStr(rotateBinlogFile));
    }

    public void heartbeatTime(String masterCode, Date timestamp) {
        HeartBeat heartBeat = HeartBeat.builder().timestamp(timestamp).masterCode(masterCode).build();
        ebWorker.send(EasyBinlogConstants.HEARTBEAT, JSONUtil.toJsonStr(heartBeat));
    }

    public void outputCount(String outputCountId, OutputResEnum outputRes) {
        OutputCount outputCount = OutputCount.builder().outputRes(outputRes).outputCountId(outputCountId).build();
        ebWorker.send(EasyBinlogConstants.OUTPUT_COUNT, JSONUtil.toJsonStr(outputCount));
    }
}